package com.cgrs572.shortlink.project.mq.consumer;

import cn.hutool.core.date.DateUtil;
import cn.hutool.core.date.Week;
import cn.hutool.core.util.StrUtil;
import cn.hutool.http.HttpUtil;
import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.cgrs572.shortlink.project.common.convention.exception.ServiceException;
import com.cgrs572.shortlink.project.dao.entity.*;
import com.cgrs572.shortlink.project.dao.mapper.*;
import com.cgrs572.shortlink.project.dto.biz.ShortLinkStatsRecordDTO;
import com.cgrs572.shortlink.project.mq.idempotent.MessageQueueIdempotentHandler;
import com.cgrs572.shortlink.project.mq.producer.DelayShortLinkStatsProducer;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RLock;
import org.redisson.api.RReadWriteLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.connection.stream.RecordId;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.stream.StreamListener;
import org.springframework.stereotype.Component;

import java.util.*;

import static com.cgrs572.shortlink.project.common.constant.RedisKeyConstant.LOCK_GID_UPDATE_KEY;
import static com.cgrs572.shortlink.project.common.constant.ShortLinkConstant.AMAP_REMOTE_URL;

/**
 * 短链接监控状态保存消息队列消费者
 */
@Slf4j
@Component
@RequiredArgsConstructor
public class ShortLinkStatsSaveConsumer implements StreamListener<String, MapRecord<String, String, String>> {

    private final ShortLinkMapper shortLinkMapper;
    private final ShortLinkGotoMapper shortLinkGotoMapper;// 用于短链接跳转
    private final RedissonClient redissonClient;// Redisson 客户端的顶层接口，可用来获取redisson分布式锁或redisson读写锁
    private final LinkAccessStatsMapper linkAccessStatsMapper; // 短链接基础访问监控持久层Bean
    private final LinkLocaleStatsMapper linkLocaleStatsMapper; // 短链接地区统计监控持久层Bean
    private final LinkOsStatsMapper linkOsStatsMapper; // 短链接访问时所用的OS统计持久层Bean
    private final LinkBrowserStatsMapper linkBrowserStatsMapper; // 短链接访问所用的浏览器统计持久层Bean
    private final LinkAccessLogsMapper linkAccessLogsMapper; // 短链接访问日志监控持久层Bean---可用来获取ip地址
    private final LinkDeviceStatsMapper linkDeviceStatsMapper; // 短链接访问设备监控持久层Bean
    private final LinkNetworkStatsMapper linkNetworkStatsMapper; // 短链接访问网络监控持久层Bean
    private final LinkStatsTodayMapper linkStatsTodayMapper; // 短链接今日访问监控持久层Bean
    private final DelayShortLinkStatsProducer delayShortLinkStatsProducer; // 延迟消费短链接统计发送者Bean
    private final StringRedisTemplate stringRedisTemplate; // 利用Java操作Redis
    private final MessageQueueIdempotentHandler messageQueueIdempotentHandler; // 消息队列幂等处理Bean

    @Value("${short-link.stats.locale.amap-key}")
    private String statsLocaleAmapKey;

    @Override
    public void onMessage(MapRecord<String, String, String> message) {
        String stream = message.getStream(); // 返回Stream的key
        RecordId id = message.getId(); // 返回消息ID

        // 判断当前消息是否已消费过
        if (!messageQueueIdempotentHandler.isMessageProcessed(id.toString())) {
            // 判断当前的这个消息流程是否执行完成
                // 加该if判断原因：若消息者在执行消息队列中未消费的消息时，突然出现极端情况导致
                // 导致执行到一半就停止了，进而导致并没有执行catch语句，
                // 为了避免已消费的消息执行到一半出现极端情况就需要进一步判断已消费的消息的整个流程是否已执行完成
                // 若执行完成则直接return，反之则直接抛出异常
            if (messageQueueIdempotentHandler.isAccomplish(id.toString())) {
                return;
            }
            throw new ServiceException("消息未完成流程，需要消息队列重试");
        }
        try {
            Map<String, String> producerMap = message.getValue(); // 返回消息内容
            String fullShortUrl = producerMap.get("fullShortUrl");
            if (StrUtil.isNotBlank(fullShortUrl)) {
                String gid = producerMap.get("gid");
                ShortLinkStatsRecordDTO statsRecord = JSON.parseObject(producerMap.get("statsRecord"), ShortLinkStatsRecordDTO.class);
                actualSaveShortLinkStats(fullShortUrl, gid, statsRecord);
            }
            // 为了避免Redis的Stream消息体越来越大，因此在消费者处理完消息队列中的对应数据后，删除在redis中的缓存
            stringRedisTemplate.opsForStream().delete(Objects.requireNonNull(stream), id.getValue());
        } catch (Throwable ex) {
            // 消费者在添加了幂等标识后的操作中执行失败了,此时就要删除缓存中的幂等标识，以免后续无法执行
            messageQueueIdempotentHandler.delMessageProcessed(id.toString());
            log.error("记录短链接监控消费异常", ex);
            throw ex; // 若不将异常抛出的话就会导致即使有异常也会执行setAccomplish
        }
        // 消费者成功消费完消息，执行成功
        messageQueueIdempotentHandler.setAccomplish(id.toString());
    }

    public void actualSaveShortLinkStats(String fullShortUrl, String gid, ShortLinkStatsRecordDTO statsRecord) {
                fullShortUrl = Optional.ofNullable(fullShortUrl).orElse(statsRecord.getFullShortUrl());
        // 前端发送请求统计短链接时,必须给其添加上Redis读写锁中的读锁---这样不会影响大量的读数据请求，但是要注意的是在读数据时是无法写数据的，适用于读多写少的情况
        RReadWriteLock readWriteLock = redissonClient.getReadWriteLock(String.format(LOCK_GID_UPDATE_KEY, fullShortUrl));
        RLock rLock = readWriteLock.readLock(); // 获取读锁（共享锁）
        // 判断是否能够添加读锁（共享锁），若不能则代表正在修改短链接相关数据
        if (!rLock.tryLock()) { // 无法添加读锁（共享锁），此时使用Redis的 延迟队列（RDelayedQueue）来让用户在指定时间后在进行统计
            delayShortLinkStatsProducer.send(statsRecord); // 发送延迟消费短链接统计
            return;
        }

        try {
            if (StrUtil.isBlank(gid)) {
                LambdaQueryWrapper<ShortLinkGotoDO> queryWrapper = Wrappers.lambdaQuery(ShortLinkGotoDO.class)
                        .eq(ShortLinkGotoDO::getFullShortUrl, fullShortUrl);
                ShortLinkGotoDO shortLinkGoToDO = shortLinkGotoMapper.selectOne(queryWrapper);
                gid = shortLinkGoToDO.getGid();
            }
            int hour = DateUtil.hour(new Date(), true); // 访问当天的时间（24小时制）
            Week week = DateUtil.dayOfWeekEnum(new Date());
            int weekValue = week.getIso8601Value(); // 访问当天是第几周
            // 短链接pv、uv、uip统计
            LinkAccessStatsDO linkAccessStatsDO = LinkAccessStatsDO.builder()
                    .pv(1)
                    .uv(statsRecord.getUvFirstFlag() ? 1 : 0)
                    .uip(statsRecord.getUipFirstFlag() ? 1 : 0)
                    .hour(hour)
                    .weekday(weekValue)
                    .fullShortUrl(fullShortUrl)
                    .gid(gid)
                    .date(new Date())
                    .build();
            linkAccessStatsMapper.shortLinkStats(linkAccessStatsDO);

            // 短链接对应所在的地区统计
                // Map集合封装请求数据
            Map<String, Object> localeParamMap = new HashMap<>();
            localeParamMap.put("key", statsLocaleAmapKey);
            localeParamMap.put("ip", statsRecord.getRemoteAddr());
                // 请求高德
            String localeResultStr = HttpUtil.get(AMAP_REMOTE_URL, localeParamMap);
                // 获取高德响应数据
            JSONObject localeResultObj = JSON.parseObject(localeResultStr);
                // 获取状态码---10000代表正确响应
            String infocode = localeResultObj.getString("infocode");
                // 获取省市信息
            String actualProvince = "未知";
            String actualCity = "未知";
            if (StrUtil.isNotBlank(infocode) && StrUtil.equals(infocode, "10000")) {
                String province = localeResultObj.getString("province");
                boolean unknownFlag = StrUtil.equals(province, "[]");
                LinkLocaleStatsDO linkLocaleStatsDO = LinkLocaleStatsDO.builder()
                        .province(actualProvince = unknownFlag ? actualProvince : province)
                        .city(actualCity = unknownFlag ? actualCity : localeResultObj.getString("city"))
                        .adcode(unknownFlag ? "未知" : localeResultObj.getString("adcode"))
                        .country("中国") // 国家标识
                        .fullShortUrl(fullShortUrl)
                        .cnt(1)
                        .gid(gid)
                        .date(new Date())
                        .build();
                linkLocaleStatsMapper.shortLinkLocaleState(linkLocaleStatsDO);
            }
            // 操作系统（即OS）统计
            LinkOsStatsDO linkOsStatsDO = LinkOsStatsDO.builder()
                    .os(statsRecord.getOs())
                    .cnt(1)
                    .gid(gid)
                    .fullShortUrl(fullShortUrl)
                    .date(new Date())
                    .build();
            linkOsStatsMapper.shortLinkOsState(linkOsStatsDO);

            // 访问浏览器统计
            LinkBrowserStatsDO linkBrowserStatsDO = LinkBrowserStatsDO.builder()
                    .browser(statsRecord.getBrowser())
                    .cnt(1)
                    .gid(gid)
                    .fullShortUrl(fullShortUrl)
                    .date(new Date())
                    .build();
            linkBrowserStatsMapper.shortLinkBrowserState(linkBrowserStatsDO);

            // 访问设备统计
            LinkDeviceStatsDO linkDeviceStatsDO = LinkDeviceStatsDO.builder()
                    .device(statsRecord.getDevice())
                    .cnt(1)
                    .gid(gid)
                    .fullShortUrl(fullShortUrl)
                    .date(new Date())
                    .build();
            linkDeviceStatsMapper.shortLinkDeviceState(linkDeviceStatsDO);

            // 访问网络统计
            LinkNetworkStatsDO linkNetworkStatsDO = LinkNetworkStatsDO.builder()
                    .network(statsRecord.getNetwork())
                    .cnt(1)
                    .gid(gid)
                    .fullShortUrl(fullShortUrl)
                    .date(new Date())
                    .build();
            linkNetworkStatsMapper.shortLinkNetworkState(linkNetworkStatsDO);

            // ip统计
            LinkAccessLogsDO linkAccessLogsDO = LinkAccessLogsDO.builder()
                    .user(statsRecord.getUv())
                    .ip(statsRecord.getRemoteAddr())
                    .browser(statsRecord.getBrowser())
                    .os(statsRecord.getOs())
                    .network(statsRecord.getNetwork())
                    .device(statsRecord.getDevice())
                    .locale(StrUtil.join("-", "中国", actualProvince, actualCity))
                    .gid(gid)
                    .fullShortUrl(fullShortUrl)
                    .build();
            linkAccessLogsMapper.insert(linkAccessLogsDO);

            // uvFirstFlag.get()---判断是否是第一次访问，若是则totalUv赋值为1,反之为0
            // 若值为true代表第一次访问，是一个新的ip；反之则不是第一次访问
            shortLinkMapper.incrementStats(gid, fullShortUrl, 1, statsRecord.getUvFirstFlag() ? 1 : 0, statsRecord.getUipFirstFlag() ? 1 : 0);
            LinkStatsTodayDO linkStatsTodayDO = LinkStatsTodayDO.builder()
                    .todayPv(1)
                    .todayUv(statsRecord.getUvFirstFlag() ? 1 : 0)
                    .todayUip(statsRecord.getUipFirstFlag() ? 1 : 0)
                    .gid(gid)
                    .fullShortUrl(fullShortUrl)
                    .date(new Date())
                    .build();
            linkStatsTodayMapper.shortLinkTodayState(linkStatsTodayDO);
        } catch (Throwable ex) {
            log.error("短链接统计异常", ex);
        } finally {
            rLock.unlock(); // 释放锁
        }
    }
}
